nit: maybe merge both switches? To avoid a copy of DmNarrow and more specifically of allRecipientIds list in DmNarrow.ofMessage.

    switch (message) {
      case StreamMessage():
        final narrow = TopicNarrow.ofMessage(message);
        locatorMap[event.message.id] = narrow;
        _addLastInStreamTopic(message.id, message.streamId, message.topic);
      case DmMessage():
        final narrow = DmNarrow.ofMessage(message, selfUserId: selfUserId);
        locatorMap[event.message.id] = narrow;
        _addLastInDm(message.id, narrow);
    }

nit:

This can be a bit more specific, I think:

Not sure the optimizer will get any useful information from that. But it seems good for matching our intent — the doc says basically the same thing a few lines above.

In principle this is a private detail of this class. It might be convenient to access it in tests; but let's make it @visibleForTesting, if not private.

It's common that a mark-read event has all its messages from the same conversation, or many messages in each of a small number of conversations. When that happens, this version will do some repeated work on those:

• looking up the list in the two maps under streams;
• more significantly, scanning through the list for the conversation to find the message to remove, and then going through the remainder of the list to move the other entries up to keep the list contiguous.

The second part in particular makes it take quadratic time to mark all the messages in a long conversation as read.

To avoid that, let's have this method first go through the list of message IDs and look them up in locatorMap, and assemble a to-do list grouped by conversation. Then it can go through that to-do list and act on each conversation just once.

For an example, see the implementation of mark-as-unread above.

(similarly for _removeAllInDms)

These checks don't fully cover the invariant that I'd want locatorMap to maintain: it should contain exactly the messages that are in streams and dms, and agree with those data structures about where each message is. These checks appear to not cover the possibility that some messages linger in locatorMap that aren't in this messages argument.

I think the cleanest way to check that would be as a separate piece of logic at the end of this function. Let's build up the whole expected locatorMap, based on the other data structures; then deepEquals to compare.